package com.genersoft.iot.vmp.service.impl;

import com.genersoft.iot.vmp.common.GeneralCallback;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.conf.security.SecurityUtils;
import com.genersoft.iot.vmp.conf.security.dto.LoginUser;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.FileUtils;
import com.genersoft.iot.vmp.utils.Md5Utils;
import com.genersoft.iot.vmp.utils.RtpPacket;
import com.genersoft.iot.vmp.utils.Threads;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.web.multipart.MultipartFile;

import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;


/**
 * @author ：mengfanlu
 * @date ：Created in 2023/8/7 16:08
 * @description：
 * @modified By：
 * @version: ${1.0.0}
 */
@Service
public class TalkbackServiceImpl implements ITalkbackService {

    private final static Logger logger = LoggerFactory.getLogger(TalkbackServiceImpl.class);

    @Autowired
    private IDeviceService deviceService;
    @Autowired
    private IMediaServerService mediaServerService;
    @Autowired
    private DynamicTask dynamicTask;
    @Autowired
    private IMediaService mediaService;
    @Autowired
    private ISIPCommander sipCommander;
    @Autowired
    private IRedisCatchStorage redisCatchStorage;
    @Autowired
    private IPlayService playService;
    @Autowired
    private IUserService userService;
    @Autowired
    private FileUtils fileUtils;
    @Autowired
    private SSRCFactory ssrcFactory;
    @Autowired
    private UserSetting userSetting;


    @Override
    public void talk(String deviceId, String channelId, String stream, String app, GeneralCallback<Object> callback) {

        checkAudioAlreadyPushWebRtc(deviceId, channelId, stream, app, (code, msg, obj) -> {
            try {
                if (code == ErrorCode.SUCCESS.getCode()) {
                    Device device = deviceService.getDevice(deviceId);
                    SendRtpItem sendRtpItem = new SendRtpItem();
                    sendRtpItem.setDeviceId(deviceId);
                    sendRtpItem.setStreamId(stream);
                    sendRtpItem.setChannelId(channelId);
                    sendRtpItem.setPlatformId(deviceId);
                    sendRtpItem.setPlayType(InviteStreamType.AUDIO);
                    sendRtpItem.setStatus(0);
                    redisCatchStorage.updateSendRTPSever(sendRtpItem);
                    //下发broadcast
                    sipCommander.audioBroadcastCmd(device, channelId);
                    String talkKey = UUID.randomUUID().toString();
                    String delayTalkKey = UUID.randomUUID().toString();

                    dynamicTask.startCron(talkKey, () -> {
                        //检测信令是否下发成功
                        SendRtpItem sendRtpItem1 = redisCatchStorage.querySendRTPServer(deviceId, channelId, stream, null, InviteStreamType.AUDIO.toString());
                        if (sendRtpItem1 == null) {
                            dynamicTask.stop(talkKey);
                            dynamicTask.stop(delayTalkKey);
                            callback.run(ErrorCode.ERROR100.getCode(), "国标对讲信令下发失败", null);
                        } else if (sendRtpItem1.getStatus() == 2) {
                            callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), null);
                            dynamicTask.stop(talkKey);
                            dynamicTask.stop(delayTalkKey);
                        }
                    }, 1000);

                    dynamicTask.startDelay(delayTalkKey, () -> {
                        //检测信令是否下发成功
                        SendRtpItem sendRtpItem1 = redisCatchStorage.querySendRTPServer(deviceId, channelId, stream, null, InviteStreamType.AUDIO.toString());

                        if (sendRtpItem1 != null && sendRtpItem1.getStatus() == 2 && dynamicTask.isAlive(talkKey)) {
                            callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), null);
                        } else {
                            callback.run(ErrorCode.ERROR100.getCode(), "国标对讲信令下发失败", null);
                            redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), sendRtpItem.getChannelId(), sendRtpItem.getCallId(), stream, InviteSessionType.AUDIO.toString());
                        }
                        dynamicTask.stop(talkKey);
                    }, 9000);
                } else {
                    callback.run(code, msg, obj);
                }
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("[国标对讲] 指令下发失败:{}", e.getMessage());
            }
        });
    }

    @Override
    public void DHTalk(String deviceId, String channelId, String stream, String app, GeneralCallback<Object> callback) {
        checkAudioAlreadyPushWebRtc(deviceId, channelId, stream, app, (code, msg, obj) -> {
            if (code == ErrorCode.SUCCESS.getCode()) {
                Device device = deviceService.getDevice(deviceId);
                MediaServerItem mediaInfo;

                if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())){
                    mediaInfo = mediaServerService.getMediaServerForMinimumLoad(null);
                } else {
                    mediaInfo = mediaServerService.getOne(device.getMediaServerId());
                }
                String mediaServerId = mediaInfo.getId();
                sipCommander.talkCmd(device,ssrcFactory.getPlaySsrc(mediaServerId),channelId,stream,app,mediaInfo);
                callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(),null);
            } else {
                callback.run(code, msg, obj);
            }
        });

    }

    @Override
    public WVPResult sendAudioFileToDevice(String deviceId, String channelId, String streamId, MultipartFile mp3File) {

        //首先判断设备是否正在对讲
        if (isAudioing(deviceId,channelId)) {
            return WVPResult.fail(ErrorCode.ERROR603);
        }
        String lock = VideoManagerConstants.BROADCAST_LOCK + deviceId;

        try {
            //1.mp3写入本地文件
            String mp3Path = fileUtils.fileWrite(mp3File.getBytes());
            //2.将mp3文件转换为pcm
            String pcmFilePath = fileUtils.mp3ToPcm(mp3Path);
            //3.创建sendrtpitem
            SendRtpItem sendRtpItem = createSendRtpItem(deviceId, channelId, streamId);
            redisCatchStorage.updateSendRTPSever(sendRtpItem);
            //4.下发国标对讲指令
            boolean lockStatus = Threads.addLock(VideoManagerConstants.BROADCAST_LOCK + deviceId);
            if (lockStatus) {
                Object localPortLock = Threads.getLock(lock);
                synchronized (localPortLock) {
                    Device device = deviceService.getDevice(deviceId);
                    sipCommander.audioBroadcastCmd(device, channelId);
                    Threads.getLock(lock).wait(5000);

                    sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, channelId, streamId, null, InviteStreamType.AUDIO.toString());

                    //5.下发数据
                    if (sendRtpItem.isTcp()) {
                        sendTCP(sendRtpItem,pcmFilePath,deviceId,channelId,streamId,device);
                    } else {
                        senUDP(sendRtpItem,pcmFilePath,deviceId,channelId,streamId,device);
                    }
            }
        }
    }catch(IOException e){
            e.printStackTrace();
        } catch(ParseException e){
            e.printStackTrace();
        } catch(InvalidArgumentException e){
            e.printStackTrace();
        } catch(SipException e){
            e.printStackTrace();
        } catch(InterruptedException e){
            e.printStackTrace();
        } finally {
            Threads.removeLock(lock);
        }
        return WVPResult.success();
    }

    @Override
    public WVPResult getWebRtcAddr(String deviceId, String channelId) {

        //首先判断设备是否正在对讲
        if (isAudioing(deviceId,channelId)) {
            return WVPResult.fail(ErrorCode.ERROR603);
        }

        Device device = deviceService.getDevice(deviceId);
        MediaServerItem mediaServerItem = playService.getNewMediaServerItem(device);

        if (mediaServerItem == null) {
            logger.error("[国标对讲] 未找到流媒体");
            return WVPResult.fail(ErrorCode.ERROR600);
        }

        Map<String, Object> result = new HashMap<>(16);
        String app = "audio";
        String stream = deviceId + "_" + channelId;
        String type = "push";
        LoginUser userInfo = SecurityUtils.getUserInfo();
        //获取推流鉴权密钥
        String sign = Md5Utils.hash(userService.getUserByUsername(userInfo.getUsername()).getPushKey());
        //示例 https://192.168.126.111:9443/index/api/webrtc?app=live&stream=test&type=play&sign=...
        String sdpIp;
        if (StringUtils.isNotEmpty(mediaServerItem.getStreamIp())) {
            sdpIp = mediaServerItem.getStreamIp();
        } else if (!ObjectUtils.isEmpty(device.getSdpIp())) {
            sdpIp = device.getSdpIp();
        } else {
            sdpIp = mediaServerItem.getSdpIp();
        }
        String webRtcPushUrl = String.format("https://%s:%s/index/api/webrtc?app=%s&stream=%s&type=%s&sign=%s", sdpIp, mediaServerItem.getHttpSSlPort(), app, stream, type, sign);
        result.put("app", app);
        result.put("stream", stream);
        result.put("type", type);
        result.put("sign", sign);
        result.put("webRtcPushUrl", webRtcPushUrl);
        logger.info("获取webrtc推流地址:{}", webRtcPushUrl);
        return WVPResult.success(result);
    }

    private void checkAudioAlreadyPushWebRtc(String deviceId, String channelId,String stream,String app, GeneralCallback<Object> callback) {
        Device device = deviceService.getDevice(deviceId);
        MediaServerItem mediaInfo;

        if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())){
            mediaInfo = mediaServerService.getMediaServerForMinimumLoad(null);
        } else {
            mediaInfo = mediaServerService.getOne(device.getMediaServerId());
        }

        String talkKey = UUID.randomUUID().toString();
        String delayTalkKey = UUID.randomUUID().toString();
        //检测webrtc是否推流成功
        dynamicTask.startCron(talkKey, ()->{
            StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaInfo.getId(), false);
            if (streamInfo != null) {
                callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), null);
                dynamicTask.stop(talkKey);
                dynamicTask.stop(delayTalkKey);
            }
        }, 1000);
        //webrtc如果3秒推送不过来就是失败
        dynamicTask.startDelay(delayTalkKey, ()->{
            StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaInfo.getId(), false);

            if (streamInfo != null && dynamicTask.isAlive(talkKey)) {
                callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), null);
            }else {
                callback.run(ErrorCode.ERROR100.getCode(), "webrtc推流超时", null);
            }
            dynamicTask.stop(talkKey);
        }, 3000);
    }

    public SendRtpItem createSendRtpItem(String deviceId,String channelId,String streamId){
        //3.下发对讲指令
        SendRtpItem sendRtpItem = new SendRtpItem();
        sendRtpItem.setDeviceId(deviceId);
        sendRtpItem.setStreamId(streamId);
        sendRtpItem.setChannelId(channelId);
        sendRtpItem.setPlatformId(deviceId);
        sendRtpItem.setPlayType(InviteStreamType.AUDIO);
        sendRtpItem.setStatus(0);
        sendRtpItem.setMp3(true);
        return sendRtpItem;
    }

    //是否正在对讲
    public boolean isAudioing(String deviceId, String channelId){
        return  redisCatchStorage.querySendRTPServer(deviceId, channelId, deviceId + "_" + channelId, null, InviteSessionType.AUDIO.toString()) == null ? false : true;
    }

    public void sendTCP(SendRtpItem sendRtpItem,String pcmFilePath,String deviceId,String channelId,String streamId,Device device){
        Thread acceptThread = new Thread(() -> {
            ServerSocket serverSocket = null;
            OutputStream os = null;
            InputStream audioStream = null;
            try {
                serverSocket = new ServerSocket(sendRtpItem.getLocalPort());
                audioStream = Files.newInputStream(Paths.get(pcmFilePath));
                int ssrc = 255;
                int audioNeedLen = 640;
                byte[] audioDataByteArray = new byte[audioNeedLen];
                RtpPacket rtp = new RtpPacket(ssrc);
                int readLineLength;
                Socket socket = serverSocket.accept();
                logger.info("开始发送语音数据,方式TCP：源 {} -> 目标 {}:{}",sendRtpItem.getLocalPort(),sendRtpItem.getIp(),sendRtpItem.getPort());
                while ((readLineLength = audioStream.read(audioDataByteArray)) != -1) {
                    byte[] payload;
                    payload = rtp.packetG711a(audioDataByteArray, audioDataByteArray.length, 14);
                    os = socket.getOutputStream();
                    os.write(payload);
                    os.flush();
                    if (readLineLength == audioNeedLen) {
                        try {
                            Thread.sleep(80);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
                // 处理客户端Socket
            } catch (IOException e) {
                // 处理异常
                e.printStackTrace();
            } finally {
                try {
                    audioStream.close();
                    os.close();
                    serverSocket.close();
                    redisCatchStorage.deleteSendRTPServer(deviceId,channelId,null,streamId,InviteStreamType.AUDIO.toString());
                    sipCommander.audioByeCmd(device,channelId,streamId, sendRtpItem.getCallId());
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (SsrcTransactionNotFoundException e) {
                    e.printStackTrace();
                } catch (InvalidArgumentException e) {
                    e.printStackTrace();
                } catch (ParseException e) {
                    e.printStackTrace();
                } catch (SipException e) {
                    e.printStackTrace();
                }

            }
        });
        acceptThread.start();
    }



    public void senUDP(SendRtpItem sendRtpItem,String pcmFilePath,String deviceId,String channelId,String streamId,Device device){
        Thread acceptThread = new Thread(() -> {
            DatagramSocket datagramSocket = null;
            InputStream audioStream = null;
            try {
                datagramSocket = new DatagramSocket(sendRtpItem.getLocalPort());
                audioStream = Files.newInputStream(Paths.get(pcmFilePath));
                int ssrc = 255;
                int audioNeedLen = 640;
                byte[] audioDataByteArray = new byte[audioNeedLen];
                RtpPacket rtp = new RtpPacket(ssrc);
                int readLineLength;
                InetAddress peerIp = InetAddress.getByName(sendRtpItem.getIp());
                logger.info("开始发送语音数据,方式UDP：源 {} -> 目标 {}:{}",sendRtpItem.getLocalPort(),sendRtpItem.getIp(),sendRtpItem.getPort());
                while ((readLineLength = audioStream.read(audioDataByteArray)) != -1) {
                    byte[] payload;
                    payload = rtp.packetG711a(audioDataByteArray, audioDataByteArray.length, 12);
                    DatagramPacket dp = new DatagramPacket(payload, payload.length, peerIp, sendRtpItem.getPort());
                    datagramSocket.send(dp);
                    if (readLineLength == audioNeedLen) {
                        try {
                            Thread.sleep(80);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
                // 处理客户端Socket
            } catch (IOException e) {
                // 处理异常
                e.printStackTrace();
            } finally {
                try {
                    audioStream.close();
                    datagramSocket.close();
                    redisCatchStorage.deleteSendRTPServer(deviceId,channelId,null,streamId,InviteStreamType.AUDIO.toString());
                    sipCommander.audioByeCmd(device,channelId,streamId, sendRtpItem.getCallId());
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (SsrcTransactionNotFoundException e) {
                    e.printStackTrace();
                } catch (InvalidArgumentException e) {
                    e.printStackTrace();
                } catch (ParseException e) {
                    e.printStackTrace();
                } catch (SipException e) {
                    e.printStackTrace();
                }

            }
        });
        acceptThread.start();
    }


}
